package gbench.appdemo.priceline.model;

import junit.framework.TestCase;

import java.text.MessageFormat;
import java.time.Duration;
import java.util.Arrays;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;

import gbench.appdemo.priceline.indicator.IIndicatorEvaluator;
import gbench.appdemo.priceline.indicator.IPriceIndicator;
import gbench.common.tree.LittleTree.IRecord;
import gbench.common.tree.LittleTree.Jdbc;
import gbench.common.tree.LittleTree.Jdbc.SqlInterceptor;

import static gbench.common.tree.LittleTree.*;
import static gbench.common.tree.LittleTree.IRecord.*;

/**
 * 图表计算模型
 * PeriodPriceLineChartModel 是一个多周期的价格线计算么模型
 *     每个周期就会拥有一个独立计算单元：processor
 * PeriodPriceLineChartModel
 * |--tradingCalendar
 * |        |--tradingDay
 * |                |--timespan
 * |--processors： 生成价格数据点的地方．
 * |            |--processor[1] --->(peroid_1)--->priceline
 * |            |            |--perister :perister就像一只蚕虫他把把价格点数据粘成(吐丝)一条线priceline存入priceLineDatabase其中．
 * |            |--processor[2] --->(peroid_2)
 * |            |--processor[i] --->(peroid_i)
 * |--priceLineDatabase
 * |            |--createTable
 * |            |--update
 * |            |--insert
 * 
 * @author xuqinghua
 *
 */
public abstract class PeriodPriceLineModel <T extends IPriceIndicator>{
    
    /**
     * 一个KLINEChart 是由一系列周期性价格线构成的。
     * @param ticker 合约代码
     * @param periods 时间间期间列表：毫秒为单位．
     */
    public PeriodPriceLineModel(String ticker,Long ...periods){
        this.primeConstruct(ticker,true, periods);//自动创建
    }
    
    /**
     * 一个KLINEChart 是由一系列周期性价格线构成的。
     * @param ticker 合约代码
     * @param auto_create 是否自动创建合约代码
     * @param periods 时间间期间列表：毫秒为单位．
     */
    public PeriodPriceLineModel(String ticker,boolean auto_create, Number ...periods){
        this.primeConstruct(ticker,auto_create, Stream.of(periods).map(Number::longValue).toArray(Long[]::new));//自动创建
    }
    
    /**
     * 价格周期计算器：把一个连续价格变动过程 给且分成 以一个个 长度为 period的 价格周期，<br>
     * 并在每个价格周期执行特定的计算任务．如果计算OHLCV则就是生成了一条Ｋ线结构．<br>
     * 
     *  PeriodPriceLineProcessor 是描述PeriodPriceLine的计算单元是一个单线程程序<br>
     *  是一个以Ｋ线节点：OHLCV为 点结构的序列．<br>
     *  PeriodPriceLineProcessor 以period 为标识，代表一种价格周期数据．<br>
     *  period 就是就是进行价格的Ｋ线统计的 时间间隔：比如，5,10,15,30等 <br>
     *  PeriodPriceLineProcessor 与 PeriodPriceLine 之间通过period 进行 关联（一一对应）<br>
     *
     *  PeriodPriceLineProcessor 就是通过一个价格周期为单位，核算出具体的 周期价格点：KLine节点的结构． <br>
     *  并给保存在LocalStorage中，LocalStorage 根据配置策略将给与同步或是异步的把数据刷新到外部存储（数据库中）<br>
     *  在价格的持续变动过程中，PeriodPriceLineProcessor 根据period 的长度，给予分解成一个一个价格的周期，每个价格<br>
     *  周期以一个OHLCV为代表（统计的价格信息），每个价格周期以一个索引index作为标识．可以通过index 在 PeriodPriceLine中<br>
     *  寻找出对应的Ｋ线节点（OHLCV) . PeriodPriceLine 就是一些列OHLCV的序列列表<br>
     *
     * @author gbench
     *
     */
    public class PeriodPriceLineProcessor {
        
        /**
         * 设置当前处理位置0
         * @param period 价格周期长度，每个价格代表一个 一个价格信息的统计范围．统计结果 
         * 如果统计是一个OHLCV结构，就是统计出了一条Ｋ线结构．
         * @param evaluator 指标计算器
         */
        public PeriodPriceLineProcessor(long period,IIndicatorEvaluator<T> evaluator) {
            this.period = period;// 价格周期长度
            this.evaluator = evaluator;
        }
        
        /**
         * 设置当前处理位置0
         * @param period 价格周期长度，每个价格代表一个 一个价格信息的统计范围．统计结果 
         * 如果统计是一个OHLCV结构，就是统计出了一条Ｋ线结构．
         * @param evaluator 指标计算器
         * @param localStorage 本地存储
         */
        public PeriodPriceLineProcessor(long period,IIndicatorEvaluator<T> evaluator,final LocalStorage<Long,IRecord> localStorage) {
            this.period = period;// 价格周期长度
            this.evaluator = evaluator;
            this.localStorage= localStorage;
        }
        
        /**
         * 设置当前处理位置0
         * @param period 价格周期长度，每个价格代表一个 一个价格信息的统计范围．统计结果 
         * 如果统计是一个OHLCV结构，就是统计出了一条Ｋ线结构．
         * @param evaluator 指标计算器
         * @param localStorage 本地存储
         * @param finder 外存查找器
         * @param persister 外存持久化器
         */
        public PeriodPriceLineProcessor(final long period,final IIndicatorEvaluator<T> evaluator,
            final Number expiredTime,final Function<Long,IRecord> finder,final Consumer<IRecord> persister) {
            
            this.period = period;// 价格周期长度
            this.evaluator = evaluator;
            this.localStorage= PeriodPriceLineProcessor.this.new LocalStorage<Long,IRecord>(expiredTime,finder,persister){
                @Override
                public PeriodPriceLineProcessor.LocalStorage<Long, IRecord> put(final Long key, final IRecord ohlc) {// 本次存储办法。
                    final var rec = this.cache.getIfPresent(key);// 提取缓存数据
                    if(rec==null) {// 新数据添加：直接存入缓存。注意此处 没有把数据持久化到数据库。因为数据会 频繁修改即写入后会被立即修改，所以在这里省略一次数据库写入。
                        this.cache.put(key, ohlc);
                    } else {// 数据更新。
                        rec.merge(ohlc,true);// 合并到rec中数据
                        if(PeriodPriceLineModel.this.syncWrite) this.persister.accept(ohlc);// 是否同步写，否则就是立即写。
                    }//if rec
                   return this;
                }// put
            };
        }
        
        /**
         * 设置存储结构
         * @param localStorage 本地存储结构
         */
        public LocalStorage<Long,IRecord> storage(final LocalStorage<Long,IRecord> localStorage) {
            this.localStorage= localStorage;
            return this.localStorage;
        }
        
        /**
         * 根据传来的价格信息
         * @param rec：价格信息的数据结构，其中有一个字段price代表即时价格
         *    --time :价格时间,
         * --price :即时价格,
         * --volume :成交量
         * @return 在时间线上刷新后的Ｋ线节点：OHLCV
         */
        public T compute(IRecord rec) {
            return this.compute(rec, this.evaluator);
        }
        
        /**
         * 根据传来的价格信息  <br>
         * @param rec：价格信息的数据结构，其中有一个字段price代表即时价格，rec 中的字段如下：  <br>
         * -- time :价格时间   <br>
         * -- price :即时价格,   <br>
         * -- volume :成交量   <br>
         * @param BiFunction<Integer,PeriodPriceLine,OHLCV> evaluator 价格计算算法：  <br>
         *  Integer 价格计算索引   <br>
         *  PeriodPriceLine 价格线条   <br>
         *  
         * @return 在时间线上刷新后的Ｋ线节点：T 类型的处理结果 一般为 一个 OHLCV 对象。
         */
        public T compute(final IRecord rec, final IIndicatorEvaluator<T> evaluator) {
            final var time = rec.date("time");// 提取价格信息中的消息数据
            if(time==null)return null;// 非法时间值给予舍弃。
            // System.out.println(PeriodPriceLineProcessor.this.period);
            final var index = dispatcher.getIndex(REC("time",rec.get("time"),"period",PeriodPriceLineProcessor.this.period));// 计算当前时段所在的K线条索引坐标．
            final var previous = (lastResult!=null && lastResult.lng("index")==index) // 同一周期且lastResult 存在
                ? lastResult // 同一周期交易结果
                : this.findOne(index);
            synchronized(evaluator) {// 同步计算 指标
                lastResult = evaluator.evaluate(rec.set("index",index),previous);// 计算K线指标
            }// 确保evaluator同步执行,以保证index不会因多线程而发生错乱
            TestCase.assertEquals(Long.valueOf(index), lastResult.lng("index"));
            this.persist(lastResult);
            final var ohlc = evaluator.getIndicator(lastResult); // 返回计算出来的K线节点
            return ohlc;
        }
        
        /**
         * 在tableName表中是否存在  ticker,period,index的数据记录
         * @param tableName 表名
         * @param ticker 合约符号
         * @param period 统计收起
         * @param index 周期序号。
         * @return 数据库中是否已经存入了ticker,period,index的数据记录。
         */
        public boolean hasOne(long index) {
            return this.findOne(index) != null;
        }

        /**
         * 在tableName表中是否存在  ticker,period,index的数据记录
         * @param tableName 表名
         * @param ticker 合约符号
         * @param period 统计收起
         * @param index 周期序号。
         * @return 数据库中 值为ticker,period,index的结果记录。
         */
        public IRecord findOne(long index) {
            return this.localStorage.get(index);
        }

        /**
         * 数据持久化
         * @param index 持久化位置:下标索引
         * @param obj 持久化对象:指标数据
         * @param mode 持久化方式：增加还是更新．
         */
        public synchronized void persist(IRecord rec) {
           this.localStorage.put(rec.lng("index"), rec);
        }
        
        /**
         * 把缓存中俄数据：刷新到数据库
         */
        public void flush() {
            this.localStorage.invalidateAll();
        }
        /**
         * 缓存状态
         * @return 状态统计
         */
        public IRecord status() {
            return this.localStorage.status();
        }

        /**
         * 文本格式化
         */
        public String toString() {
            return MFT("pplcm-{0}-{1,number,#}",ticker,this.period); // // pplcm 就是PeriodPriceLineChartModel的缩写
        }
        
        /**
         * 这是一个抽象类 用于 定义一个简单数据存储
         * 数据货栈:之所以叫LocalStorage 是为了避免与 guava的LocalCache相一致
         * @author gbench
         *
         */
        public abstract class LocalStorage<K,V>{
            /**
             * 本地缓存的构造函数
             * @param expiredTime 过期时间 单位毫秒
             */
            public LocalStorage(final Number expiredTime,final Function<K,V> finder,final Consumer<V> persister){
                
                this.expiredTime = expiredTime.longValue();
                this.persister = persister;
                this.finder = finder;
                
                // 把过期的数据写入数据库
                final RemovalListener<K, V> listener = (RemovalNotification<K,V> notification)->{
                    if(!PeriodPriceLineModel.this.syncWrite) { // 仅当为非同步写的情形才做根据过期时间进行数据写入。
                        final var ohlc = notification.getValue();// 提取数据项目
                        System.out.println("expired to write:"+ohlc);
                        // 需要persister进行持久化，不然会出现不同时间周期的数据保存混乱的情况
                        this.persister.accept(ohlc);// 把ohlc 数据进行持久化
                    }// 同步写入检查。
                };// listener
                
                cache = CacheBuilder.newBuilder()// 数据缓存
                    .initialCapacity(5) //设置cache的初始大小为10，要合理设置该值 
                    .concurrencyLevel(1) //设置并发数为1，即同一时间最多只能有5个线程往cache执行写入操作,因为 processor 是单线程程序
                    .expireAfterWrite(Duration.ofMillis(this.expiredTime)) //设置cache中的数据在写入之后的存活时间为10秒
                    //.recordStats()// 记录统计数据
                    .removalListener(listener)
                    .build(new CacheLoader<K,V>(){
                        public V load (final K key){
                            return LocalStorage.this.finder.apply(key);
                        }// load
                    });  //构建cache实例
                
            }// LocalCache
            
            /**
             * 读取本地数据
             * @param key 键名
             * @param value 键值
             * @return V 的对象
             */
            public V get(final K key) {
                return cache.getIfPresent(key);
            }
            
            /**
             * 把value 以key的名义(为键值) 存入本地存储。
             * @param key 键名
             * @param value 键值
             * @return 本地存储本身，以方便链式编程。
             */
            public abstract LocalStorage<K,V> put(final Long key, final V value);
            
            /**
             * 缓存状态统计
             * @return 缓存状态统计
             */
            public IRecord status() {
                return REC("size",this.cache.size(),"stats",this.cache.stats(),
                    "data",cache.asMap());
            }
            
            /**
             * 把所有数据给予清空
             */
            public void invalidateAll() {
                this.cache.invalidateAll();
            }
            
            protected final Consumer<V> persister;
            protected final Function<K,V> finder;
            protected final long expiredTime;// 过期时间
            protected final Cache<K,V> cache; // 数据缓存
            
        }
        
        /**
         * 处理器的处理周期
         * @return 处理器的处理周期
         */
        public long getPeriod() {
            return period;
        }
        
        /**
         * 合约代码
         * @return 合约代码
         */
        public String getTicker() {
            return PeriodPriceLineModel.this.ticker;
        }
        
        /**
         * 获取估算器
         * @return 估算器
         */
        public IIndicatorEvaluator<T> getEvaluator() {
            return this.evaluator;
        }
        
        private long period; //价格周期
        private IRecord lastResult = null; // 最后的计算结果
        private IIndicatorEvaluator<T> evaluator;// 指标计算器
        
        /**
         *  采用priceLineDatabase 作为持久化媒介的 本地存储。 
         */
        private LocalStorage<Long,IRecord> localStorage;
        
    } // PeriodPriceLineProcessor
    
    /**
     * 基本构造：PeriodPriceLineChartModel 的各个属性对象。
     * 一个 PeriodPriceLineChartModel 是由一系列周期性价格线构成的。<br>
     * @param ticker 合约代码
     * @param auto_create 是否自动创建合约代码:即为不存在的合约自动创建一张表
     * @param periods 时间间期间列表：毫秒为单位．
     */
    private void primeConstruct(final String ticker, final boolean auto_create,final Long ...periods){
        this.ticker = ticker;// 价格线模型所跟踪的 合约代码
        this.auto_create = auto_create;// 设置价格线数据表的自动创建标记。
        Stream.of(periods).forEach(this::addProcessor);// 生成K线生成器：每个时间间期period都是一个处理器
    }

    /**
     * 添加一个价格处理器，进行OHLCV处理
     * @param period 价格周期
     */
    public PeriodPriceLineModel<T> addProcessor(final long period) {
        if(processors==null) processors= new LinkedList<>();
        if(processors.stream().filter(e->e.period==period).findAny().isPresent()) {
            System.out.println("重复周期不予添加");
            return this;
        }
        
        final var tblName = PeriodPriceLineModel.this.getTableName();
        //添加一条周期价线条
        processors.add(new PeriodPriceLineProcessor(period,ohlc_evaluator,
            PeriodPriceLineModel.this.processorStorageExpiredTime,
            index -> priceLineDatabase.findOne(tblName, ticker, period, index ), // finder
            ohlc -> priceLineDatabase.persist(tblName, ticker, period, // persister
                ohlc.lng("index"),ohlc.date("time"),
                ohlc.dbl("open"),ohlc.dbl("high"), ohlc.dbl("low"), ohlc.dbl("close"),
                ohlc.lng("volume")- ohlc.lng("volume0") ,ohlc.lng("volume0"))
            ));// processors.add
        
        return this;
    }

    /**
     * 价格信息的处理
     * @param rec 价格信息
     */
    public void handle(final IRecord rec) {
        handle(rec,this.callback);
    }
    
    /**
     * 价格信息的处理：工艺流程的处理过程。
     * @param rec 价格信息
     * @param callback 结果处理的回调函数:(result)->{ 插件深加工reprocess的处理逻辑},
     * @param result 就是processor 生成的结果记录  其中 result 中含有一个字段：processor 用于标识
     * result的制作工艺：即result的类型。对于kline 的生成工艺：processor 是
     * pplcm-{ticker}-{period}的字段结构，花括号表示变量字段。PPLCM 就是PeriodPriceLineChartModel
     */
    public void handle(final IRecord rec,final BiConsumer<PeriodPriceLineProcessor,T>callback) {
        final BiConsumer<PeriodPriceLineProcessor,T> cb = (callback==null) ? ((p,r)->{}) : (callback);// 回调函数：提供一个默认的回调函数。
        // 依次处理各个价格信息．
        processors.parallelStream().forEach(processor->{ try { // 把处理函数包装再try-catch之中。
            final var res = processor.compute(rec,this.ohlc_evaluator);// 计算指标数据:这就是出发计算K线的开始位置：梦开始的地方。！！！
            if(res==null) {
                System.err.println(MFT("处理器:{1} 处理  {0} 时生成的计算结果 为null，给与舍弃！",rec,processor));
                return;
            }
            if(cb!=null)cb.accept(processor,res);// 对处理后生成的结果 给与回调函数。
        } catch (Exception e) {e.printStackTrace(); } } );// processors.parallelStream
    }
    
    /**
     * 初始化：图表结构
     * jdbcConfig: 数据库连接配置  默认为 null<br>
     * dispatcher: 分发器 默认为 null <br>
     * evaluator: 核算器  默认为 null<br>
     * callback: 结果 回调  默认为 null<br>
     * sync_write: 同步写入 默认为  false  <br>
     * rclient: redisclient 默认为 null<br>
     * processor_storage_expired_time: 处理器的缓存过期时间 ,单位毫秒,默认一分钟 <br>
     * priceline_refresh_interval: 价格线缓存刷新间,单位毫秒,隔 默认5秒 <br>
     * @param cfg 参数配置,配置项目详见 上文注释
     */
    @SuppressWarnings("unchecked")
    public void initialize(final IRecord cfg) {
        final var final_cfg = cfg==null?REC():cfg;
        final var jdbcConfig = (Map<String,String>)(Object)final_cfg.asMap("jdbc");
        this.jdbcInit(jdbcConfig);// 数据库jdbc配置
        final var dispatcher = final_cfg.as("dispatcher",IDispatcher.class);
        this.dispatcher(dispatcher);// 分发器
        final var evaluator = (IIndicatorEvaluator<T>)final_cfg.as("evaluator",IIndicatorEvaluator.class);
        this.evaluator(evaluator);// 核算器
        final var cb = (BiConsumer<PeriodPriceLineProcessor,T>)final_cfg.as("callback",Consumer.class);
        this.callback(cb);// 结果回调
        this.setSyncWrite(final_cfg.bool("sync_write",false));// 是否同步写入
        this.rclient = final_cfg.as("rclient",IRedisClient.class);
        this.processorStorageExpiredTime = final_cfg.lng("processor_storage_expired_time",this.processorStorageExpiredTime);
        this.pricelineRefreshInterval = final_cfg.lng("priceline_refresh_interval",this.processorStorageExpiredTime);
    }

    /**
     * 把缓存数据刷新到数据库
     */
    public void flush() {
        this.processors.forEach(PeriodPriceLineProcessor::flush);
    }
    
    /**
     * 把缓存数据刷新到数据库
     */
    public List<IRecord> status() {
        return this.processors.stream().map(e->e.status()).collect(Collectors.toList());
    }
    
    /**
     * 初始化：图表结构
     * @param jdbcConfig jdbc的数据配置,jdbcConfig 可以为空，代表使用默认配置．
     * driver,url,user,password
     */
    public void jdbcInit(final Map<String,String> jdbcConfig) {
        
        if(jdbcConfig!=null && jdbcConfig.size()>0) {// 指定特定的数据库
            //  初始化价格线数据库
            priceLineDatabase = Jdbc.newInstance(PriceLineDatabase.class,
                jdbcConfig,null,pldb_interceptor());// 方法拦截：以便利用缓存。
            System.out.println("为"+this+"配置价格线的数据库为:"+jdbcConfig);
        } else {
            System.out.println(this.getClass().getName()+"采用默认的价格线数据库：该情形应该只出现在测试环境！！！");
        }
        
        synchronized (this.getClass()){// 创建一个临界区
            final boolean b = priceLineDatabase.exists(this.getTableName());
            if(!b && auto_create) {// 表格不存在时则自动创建
                priceLineDatabase.createTable(this.getTableName());
            }//if 表格是否存在
        }// 创建一个临界区
    }
    
    /**
     * PriceLine DataBase interceptor (pldb interceptor)
     * 方法拦截器：拦截方法 intercept 返回null 继续运行本方法,非空直接返回。
     * @return SqlInterceptor SQL 语句的执行拦截器
     */
    public SqlInterceptor<List<IRecord>> pldb_interceptor() {
        // p:pattern,a:args
        final BiFunction<String,IRecord,String> tosql = (pattern,params) -> MFT(pattern,params.toObjArray());
        final var prefix = "pplcm:intercept";// intercept 方法 通用前缀
        
        // 生成一个方法拦截器
        return (m,a,p,j)->{// 方法前置拦截器:Method method,IRecord params,String sqlpattern,Jdbc jdbc
            if(this.rclient==null)return null;// 返回继续执行。如果没有设置redis不予进行拦截。
            final String methodName = m.getName();// 方法对象的名称，即调用方法
            List<IRecord> recs = null;// 返回值
            final Function<String,Supplier<List<IRecord>>> initiator = (label)->()->{// 注意这是一个二元函数
                final var sql = tosql.apply(p, a);
                //System.out.println(MFT("{0}@{1}缓存初始化:{2}",tid(),label,sql));
                return j.sql2records(sql);
            };// 初始化函数
            
            switch(methodName) {// 在方法名的层级 上进行 拦截。
                case "findOne" :{// 查询函数
                    recs = rclient.rll( concat2key(prefix,methodName,a.get("ticker"),a.get("period"),a.get("index")),5000,
                        initiator.apply(methodName));
                    break;
                }// findOne
                
                case "exists" : {// 表数据是否存在
                    final String key = concat2key(prefix,methodName,getTableName());
                    recs = rclient.rll(key,3600000,initiator.apply(methodName));
                    break;
                }// exists
                
                case "getPriceline" : {// 表数据是否存在
                    if(a.lng("size")>1000)return null;// 对于超长的数据不予缓存。
                    final String args = Arrays.stream(m.getParameters())
                        .map(e-> //e.getType().getSimpleName()+" "+
                            e.getName().substring(0,1))// 类型名参数
                        .collect(Collectors.joining(""));// 注意分割符号需要与下面的signature保持一致。
                    final String key = concat2key(prefix,methodName,args,getTableName(),
                        a.get("ticker"),a.get("period"),a.get("start"),a.get("size"));// 缓存访问的键值。
                    //System.out.println(concat2key(new Date(),"--->",key));
                    //final var signature="String tableName,String ticker,long period,long start,long size";// 参数全签名列表
                    //final var signature = "tableName,ticker,period,start,size";// 参数全签名列表
                    final var signature = "ttpss";// 参数签名缩写列表,用一个简单一点儿的方法列表
                    if(!args.equals(signature))return null;// 只对tableName,ticker,period,start,size的方法参数进行数据缓存
                    final var timeout = this.getPricelineRefreshInterval();
                    recs = rclient.rll(key,(timeout>0?timeout:5000),initiator.apply(methodName));
                    break;
                }// getPriceline
                
                default:{
                    //final var sql = tosql.apply(p, a);
                    //System.out.println(tid()+":"+m.getName()+"--->"+sql);
                }
            }// switch
            
            return recs;
        };
    }
    
    /**
     * 获得表格名称:名称均为小写形式。
     * 采用"t_price_line_"+合约符号名tikcer作为表名,比如对于MA005的甲醇2020年的五月份的合约
     * 该合约数据的对应的价格线的数据表名为:t_price_line_ma005
     * @return 价格线的数据表名
     */
    public String getTableName() {
        final String tbl="t_price_line";// 默认前缀
        return single?tbl:tbl+"_"+ticker.toLowerCase();// 多表存数据
    }
    
    /**
     * 清除所有：数据价格数据记录
     */
    public void delete(final String ticker) {
        priceLineDatabase.delete(getTableName(), ticker);
    }

    /**
     * 清除所有：数据价格数据记录
     */
    public void clearAll() {
        priceLineDatabase.dropTable(this.getTableName());
        priceLineDatabase.createTable(this.getTableName());
        priceLineDatabase.createIndex(this.getTableName());
    }

    /**
     * ticker对应的价格线是否有效
     * @return true 价格线的数据表存在， false 价格线的数据表不存在，
     */
    public boolean isPriceLineAvailable(){
        return priceLineDatabase.exists(this.getTableName());
    }
    
    /**
     *   返回隶属于period的ticker合约代码在index处的kpoint,
     * @param ticker 合约代码
     * @param period 时间周期，60000一分钟
     * @param index 所有下标
     * @return IRecord
     */
    public IRecord getKPoint(String ticker, long period, long index){
        return priceLineDatabase.getKPoint(this.getTableName(), ticker, period, index);
    }
    
    /**
     *  所有期货数据存入一张表
     */
    public void enableSingleTable() {
        single = true;
    }
    
    /**
     *  所有期货数据存入一张表
     */
    public void disableMultiTable() {
        single = true;
    }
    
    /**
     * K线信息的内置刷新频率,单位毫秒 :
     * getPriceLine 方法的刷新周期。
     * @return
     */
    public long getPricelineRefreshInterval() {
        return pricelineRefreshInterval;
    }

    /**
     * K线信息的内置刷新频率,单位毫秒
     * @param pricelineRefreshInterval
     */
    public void setPricelineRefreshInterval(final long pricelineRefreshInterval) {
        this.pricelineRefreshInterval = pricelineRefreshInterval;
    }
    
    /**
     * 格式化数据输出
     */
    public String toString() {
        return "KlineChart-"+ticker;
    }

    /**
     * 参数连接 成为一个redis key
     * @param objs 对象列表
     * @return 把objs根据其值类型串联起来的字符串
     */
    private static String concat2key(final Object...objs) {
        final var s = Stream.iterate(0,i->i<objs.length,i->i+1)
        .map(i->{
            String line = null;
            if(objs[i] instanceof Date) {
                line = MFT("{0},date,yyyMMddHHmmss",i);
            } else if(objs[i] instanceof Number) {
                line = MFT("{0},number,#",i);
            } else {
                line = MFT("{0}",i);
            }
            return "{"+line+"}";
        }).collect(Collectors.joining(":"));
        return MessageFormat.format(s,objs);
    }

    /**
     * K 线数据是否同步写入数据库
     * @return 是否同步写入数据库
     */
    public boolean isSyncWrite() {
        return syncWrite;
    }

    /**
     * K 线数据是否同步写入数据库
     * @param syncWrite 同步写入的标记
     */
    public void setSyncWrite(boolean syncWrite) {
        this.syncWrite = syncWrite;
    }
    
    public long getProcessorStorageExpiredTime() {
        return processorStorageExpiredTime;
    }

    public void setProcessorStorageExpiredTime(long processorStorageExpiredTime) {
        this.processorStorageExpiredTime = processorStorageExpiredTime;
    }
    
    /**
     * 分发器
     * @param dispatcher 分发器
     * @return IDispatcher
     */
    public IDispatcher dispatcher(IDispatcher dispatcher) {
        this.dispatcher = dispatcher;
        return this.dispatcher;
    }
    
    /**
     * 指标计算器
     * @param ohlc_evaluator 指标计算器
     * @return IIndicatorEvaluator&lt;T&gt;
     */
    public IIndicatorEvaluator<T> evaluator(final IIndicatorEvaluator<T> ohlc_evaluator) {
        this.ohlc_evaluator = ohlc_evaluator;
        return this.ohlc_evaluator;
    }
    
    /**
     * 结果回调函数
     * @param callback 结果回调函数:(processor,result->{}
     * @return BiConsumer&lt;PeriodPriceLineProcessor,T&gt;
     */
    public BiConsumer<PeriodPriceLineProcessor,T> callback(final BiConsumer<PeriodPriceLineProcessor,T> callback) {
        this.callback = callback;
        return this.callback;
    }
    
    /**
     * 读取价格线
     * @param tableName 表名
     * @param ticker 合约代码
     * @param period 统计周期
     * @param start 开始索引
     * @param size 最大长度
     * @return 价格线信息
     */
    public List<IRecord> getPriceline( final String tableName, final String ticker, 
            final long period, final long start, final long size ){
        return this.priceLineDatabase.getPriceline(tableName, ticker, period, start, size);
    }
    
    /**
     * 读取价格线
     * @param tableName 表名
     * @param ticker 合约代码
     * @param period 统计周期
     * @param start 开始索引
     * @param size 最大长度
     * @return 价格线信息
     */
    public List<IRecord> getPriceline( final String ticker, 
            final long period, final long start, final long size ){
        return this.getPriceline(this.getTableName(), ticker, period, start, size);
    }
    
    protected Date startTime = null;// 开始时间
    protected Date endTime = null;// 结束时间
    protected String ticker = null;// 期货代码,此为AbstractChartModel的标识属性.
    protected boolean single = false;// 所有期货数据是否存入一张表
    private boolean syncWrite = true; // 缓存数据是否同步写入数据库。异步写入数据会比较快，但是由于是依据是异步写入就会出现前端的读取价格线(K线)数据不及时的情况
    protected boolean auto_create = true;// 自动创建价格线的数据表。
    protected long pricelineRefreshInterval = 5000;// K线信息的内置刷新频率,单位毫秒
    protected long processorStorageExpiredTime = 60l*1000;// 处理器的缓存过期时间
    protected PriceLineDatabase priceLineDatabase = Jdbc.newInstance(PriceLineDatabase.class);// 价格线段处理器
    protected List<PeriodPriceLineProcessor> processors = new LinkedList<>();// 周期价格线处理器
    protected IDispatcher dispatcher;
    protected IRedisClient rclient;
    protected BiConsumer<PeriodPriceLineProcessor,T> callback;// 结果回调
    protected IIndicatorEvaluator<T> ohlc_evaluator; // OHLC计算器
    public static final Pattern digits = Pattern.compile("\\s*(\\d+)\\s*");// 纯数字
    public static final Pattern digits_unit = Pattern.compile("\\s*(\\d+)\\s*([yMwdhms])\\s*");//数字加单位
    public static boolean debug = false;// 调试标记
}